package org.topteam.esb.monitor;

import org.apache.log4j.Logger;
import org.topteam.esb.akka.AkkaEsbSystem;
import org.topteam.esb.monitor.MonitorEvent.ConsumerCallFailureEvent;
import org.topteam.esb.monitor.MonitorEvent.NodeSeedAllEvent;
import org.topteam.esb.monitor.MonitorEvent.NodeSeedJionEvent;
import org.topteam.esb.monitor.MonitorEvent.NodeSeedLeaveEvent;
import org.topteam.esb.monitor.MonitorEvent.ProviderBeCalledCount;

import akka.actor.ActorRef;
import akka.actor.UntypedActor;
import akka.contrib.pattern.DistributedPubSubExtension;
import akka.contrib.pattern.DistributedPubSubMediator;

public class MonitorClientActor extends UntypedActor {
	private Logger logger = Logger.getLogger(MonitorClientActor.class);

	private ActorRef mediator = DistributedPubSubExtension.get(
			getContext().system()).mediator();

	public MonitorClientActor(AkkaEsbSystem akkaEsbSystem) {
		mediator.tell(new DistributedPubSubMediator.Subscribe(
				MonitorEvent.MONITOR_PASSIVE_EVENT_TOPIC, getSelf()), getSelf());
		logger.info("启动监控客户端服务成功。。。");
	}

	@Override
	public void onReceive(Object message) throws Exception {
		if (message instanceof ConsumerCallFailureEvent) {
			ConsumerCallFailureEvent event = (ConsumerCallFailureEvent) message;
			mediator.tell(new DistributedPubSubMediator.Publish(
					MonitorEvent.MONITOR_EVENT_TOPIC, event), getSelf());
		} else if (message instanceof ProviderBeCalledCount) {
			ProviderBeCalledCount event = (ProviderBeCalledCount) message;
			mediator.tell(new DistributedPubSubMediator.Publish(
					MonitorEvent.MONITOR_EVENT_TOPIC, event), getSelf());
		} else if (message instanceof NodeSeedJionEvent) {
			NodeSeedJionEvent event = (NodeSeedJionEvent) message;
			mediator.tell(new DistributedPubSubMediator.Publish(
					MonitorEvent.MONITOR_EVENT_TOPIC, event), getSelf());
		} else if (message instanceof NodeSeedLeaveEvent) {
			NodeSeedLeaveEvent event = (NodeSeedLeaveEvent) message;
			mediator.tell(new DistributedPubSubMediator.Publish(
					MonitorEvent.MONITOR_EVENT_TOPIC, event), getSelf());
		} else if (message instanceof NodeSeedAllEvent) {
			NodeSeedAllEvent event = (NodeSeedAllEvent) message;
			mediator.tell(new DistributedPubSubMediator.Publish(
					MonitorEvent.MONITOR_EVENT_TOPIC, event), getSelf());
		} else {
			unhandled(message);
		}
	}

}
